package com.sia.springcloud.process.consumer;

import com.alibaba.fastjson.JSONObject;
import com.sia.springcloud.process.influxDB.InfluxDbTool;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @Author: Sirius
 * @Date: 2022/5/14 17:39
 * @Description:
 **/
public class KafkaConsumerBosh implements Runnable {



    private KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private InfluxDbTool influxDbTool;
    private String topic;
    private static final String GROUPID = "etl";


    public KafkaConsumerBosh(String topicName, InfluxDbTool influxDbTool) {
        this.topic = topicName;
        this.influxDbTool = influxDbTool;
        init();
    }

    @Override
    public void run() {
        System.out.println("---------开始消费---------");
        List<String> list = new ArrayList<String>();
        List<Long> list2 = new ArrayList<Long>();
        BatchPoints batchPoints = BatchPoints.builder().build();
        while (true) {
            try {
                msgList = consumer.poll(100);
                if (null != msgList) {
                    for (ConsumerRecord<String, String> record : msgList) {
                        list.add(record.value());
                        list2.add(record.offset());
                        if (record.topic().equals("bosh") && null != record && record.value().contains("{")) {
                            JSONObject jsonObject = JSONObject.parseObject(record.value());
                            /**
                             * 设备名作为表名，字段名作为tag
                             */

                            Point point = Point.measurement(jsonObject.getString("timerName"))
                                    //sql
                                    .addField("id", jsonObject.getString("id"))
                                    .tag("progNo", jsonObject.getString("progNo"))
                                    .tag("protRecord_ID", jsonObject.getString("protRecord_ID"))
                                    .tag("dateTime", jsonObject.getString("dateTime"))
                                    .tag("spotName", jsonObject.getString("spotName"))
                                    .tag("wear", jsonObject.getString("wear"))
                                    .tag("wearPerCent", jsonObject.getString("wearPerCent"))
                                    .tag("monitorState", jsonObject.getString("monitorState"))
                                    .tag("monitorState_txt", jsonObject.getString("monitorState_txt"))
                                    .tag("regulationState", jsonObject.getString("regulationState"))
                                    .tag("regulationState_txt", jsonObject.getString("regulationState_txt"))
                                    .tag("measureState", jsonObject.getString("measureState"))
                                    .tag("measureState_txt", jsonObject.getString("measureState_txt"))
                                    .tag("powerState", jsonObject.getString("powerState"))
                                    .tag("powerState_txt", jsonObject.getString("powerState_txt"))
                                    .tag("sequenceState", jsonObject.getString("sequenceState"))
                                    .tag("sequenceState_txt", jsonObject.getString("sequenceState_txt"))
                                    .tag("sequenceStateAdd", jsonObject.getString("sequenceStateAdd"))
                                    .tag("sequenceStateAdd_txt", jsonObject.getString("sequenceStateAdd_txt"))
                                    .tag("sequenceRepeat", jsonObject.getString("sequenceRepeat"))
                                    .tag("sequenceRepeat_txt", jsonObject.getString("sequenceRepeat_txt"))
                                    .tag("monitorMode", jsonObject.getString("monitorMode"))
                                    .tag("monitorMode_txt", jsonObject.getString("monitorMode_txt"))
                                    .tag("iDemandStd", jsonObject.getString("iDemandStd"))
                                    .tag("ilsts", jsonObject.getString("ilsts"))
                                    .tag("regulationStd", jsonObject.getString("regulationStd"))
                                    .tag("regulationStd_txt", jsonObject.getString("regulationStd_txt"))
                                    .tag("iDemand1", jsonObject.getString("iDemand1"))
                                    .tag("iActual1", jsonObject.getString("iActual1"))
                                    .tag("regulation1", jsonObject.getString("regulation1"))
                                    .tag("regulation1_txt", jsonObject.getString("regulation1_txt"))
                                    .tag("iDemand2", jsonObject.getString("iDemand2"))
                                    .tag("iActual2", jsonObject.getString("iActual2"))
                                    .tag("regulation2", jsonObject.getString("regulation2"))
                                    .tag("regulation2_txt", jsonObject.getString("regulation2_txt"))
                                    .tag("iDemand3", jsonObject.getString("iDemand3"))
                                    .tag("iActual3", jsonObject.getString("iActual3"))
                                    .tag("regulation3", jsonObject.getString("regulation3"))
                                    .tag("regulation3_txt", jsonObject.getString("regulation3_txt"))
                                    .tag("phaStd", jsonObject.getString("phaStd"))
                                    .tag("pha1", jsonObject.getString("pha1"))
                                    .tag("pha2", jsonObject.getString("pha2"))
                                    .tag("pha3", jsonObject.getString("pha3"))
                                    .tag("t_iDemandStd", jsonObject.getString("t_iDemandStd"))
                                    .tag("tActualStd", jsonObject.getString("tActualStd"))
                                    .tag("partIdentString", jsonObject.getString("partIdentString"))
                                    .tag("tipDressCounter", jsonObject.getString("tipDressCounter"))
                                    .tag("electrodeNo", jsonObject.getString("electrodeNo"))
                                    .tag("voltageActualValue", jsonObject.getString("voltageActualValue"))
                                    .tag("voltageRefValue", jsonObject.getString("voltageRefValue"))
                                    .tag("currentActualValue", jsonObject.getString("currentActualValue"))
                                    .tag("currentReferenceValue", jsonObject.getString("currentReferenceValue"))
                                    .tag("weldTimeActualValue", jsonObject.getString("weldTimeActualValue"))
                                    .tag("weldTimeRefValue", jsonObject.getString("weldTimeRefValue"))
                                    .tag("energyActualValue", jsonObject.getString("energyActualValue"))
                                    .tag("energyRefValue", jsonObject.getString("energyRefValue"))
                                    .tag("powerActualValue", jsonObject.getString("powerActualValue"))
                                    .tag("powerRefValue", jsonObject.getString("powerRefValue"))
                                    .tag("resistanceActualValue", jsonObject.getString("resistanceActualValue"))
                                    .tag("resistanceRefValue", jsonObject.getString("resistanceRefValue"))
                                    .tag("pulseWidthActualValue", jsonObject.getString("pulseWidthActualValue"))
                                    .tag("pulseWidthRefValue", jsonObject.getString("pulseWidthRefValue"))
                                    .tag("stabilisationFactorActValue", jsonObject.getString("stabilisationFactorActValue"))
                                    .tag("stabilisationFactorRefValue", jsonObject.getString("stabilisationFactorRefValue"))
                                    .tag("thresholdStabilisationFactor", jsonObject.getString("thresholdStabilisationFactor"))
                                    .tag("wldEffectStabilisationFactor", jsonObject.getString("wldEffectStabilisationFactor"))
                                    .tag("uipActualValue", jsonObject.getString("uipActualValue"))
                                    .tag("uipRefValue", jsonObject.getString("uipRefValue"))
                                    .tag("uirExpulsionTime", jsonObject.getString("uirExpulsionTime"))
                                    .tag("uirMeasuringActive", jsonObject.getString("uirMeasuringActive"))
                                    .tag("uirMeasuringActive_txt", jsonObject.getString("uirMeasuringActive_txt"))
                                    .tag("uirRegulationActive", jsonObject.getString("uirRegulationActive"))
                                    .tag("uirRegulationActive_txt", jsonObject.getString("uirRegulationActive_txt"))
                                    .tag("uirMonitoringActive", jsonObject.getString("uirMonitoringActive"))
                                    .tag("uirMonitoringActive_txt", jsonObject.getString("uirMonitoringActive_txt"))
                                    .tag("uirWeldTimeProlongationActive", jsonObject.getString("uirWeldTimeProlongationActive"))
                                    .tag("uirWeldTimeProlongationActive_txt", jsonObject.getString("uirWeldTimeProlongationActive_txt"))
                                    .tag("uirQStoppRefCntValue", jsonObject.getString("uirQStoppRefCntValue"))
                                    .tag("uirQStoppActCntValue", jsonObject.getString("uirQStoppActCntValue"))
                                    .tag("uirUipUpperTol", jsonObject.getString("uirUipUpperTol"))
                                    .tag("uirUipLowerTol", jsonObject.getString("uirUipLowerTol"))
                                    .tag("uirUipCondTol", jsonObject.getString("uirUipCondTol"))
                                    .tag("uirPsfLowerTol", jsonObject.getString("uirPsfLowerTol"))
                                    .tag("uirPsfCondTol", jsonObject.getString("uirPsfCondTol"))
                                    //rui
                                    .tag("VERSION_NO", jsonObject.getString("vERSION_NO"))
                                    .tag("DAY", jsonObject.getString("dAY"))
                                    .tag("MONTH", jsonObject.getString("mONTH"))
                                    .tag("YEAR", jsonObject.getString("yEAR"))
                                    .tag("HOUR", jsonObject.getString("hOUR"))
                                    .tag("MINUTE", jsonObject.getString("mINUTE"))
                                    .tag("SECOND", jsonObject.getString("sECOND"))
                                    .tag("USER_COMMENT", jsonObject.getString("uSER_COMMENT"))
                                    .tag("PROGRAM_NUMBER", jsonObject.getString("pROGRAM_NUMBER"))
                                    .tag("SOFTWARE_VERSION", jsonObject.getString("sOFTWARE_VERSION"))
                                    .tag("NUMBER_OF_DATA", jsonObject.getString("nUMBER_OF_DATA"))
                                    .tag("UMAX", jsonObject.getString("uMAX"))
                                    .tag("IMAX", jsonObject.getString("iMAX"))
                                    .tag("ENERGIE", jsonObject.getString("eNERGIE"))
                                    .tag("PMAX", jsonObject.getString("pMAX"))
                                    .tag("CUR_CORR_FACTOR", jsonObject.getString("cUR_CORR_FACTOR"))
                                    .tag("SPLASH_TIME", jsonObject.getString("sPLASH_TIME"))
                                    .tag("WEAR_COUNTER", jsonObject.getString("wEAR_COUNTER"))
                                    .tag("MILLING_COUNTER", jsonObject.getString("mILLING_COUNTER"))
                                    .tag("GUN_RESISTANCE", jsonObject.getString("gUN_RESISTANCE"))
                                    .tag("ZMAX", jsonObject.getString("zMAX"))
                                    .tag("DATA_SET1_1", jsonObject.getString("dATA_SET1_1"))
                                    .tag("DATA_SET2_1", jsonObject.getString("dATA_SET2_1"))
                                    .tag("DATA_SET1_2", jsonObject.getString("dATA_SET1_2"))
                                    .tag("DATA_SET2_2", jsonObject.getString("dATA_SET2_2"))
                                    .tag("DATA_SET1_3", jsonObject.getString("dATA_SET1_3"))
                                    .tag("DATA_SET2_3", jsonObject.getString("dATA_SET2_3"))
                                    .tag("DATA_SET1_4", jsonObject.getString("dATA_SET1_4"))
                                    .tag("DATA_SET2_4", jsonObject.getString("dATA_SET2_4"))
                                    .tag("DATA_SET1_5", jsonObject.getString("dATA_SET1_5"))
                                    .tag("DATA_SET2_5", jsonObject.getString("dATA_SET2_5"))
                                    .tag("DATA_SET1_6", jsonObject.getString("dATA_SET1_6"))
                                    .tag("DATA_SET2_6", jsonObject.getString("dATA_SET2_6"))
                                    .tag("DATA_SET1_7", jsonObject.getString("dATA_SET1_7"))
                                    .tag("DATA_SET2_7", jsonObject.getString("dATA_SET2_7"))
                                    .tag("DATA_SET1_8", jsonObject.getString("dATA_SET1_8"))
                                    .tag("DATA_SET2_8", jsonObject.getString("dATA_SET2_8"))
                                    .tag("DATA_SET1_9", jsonObject.getString("dATA_SET1_9"))
                                    .tag("DATA_SET2_9", jsonObject.getString("dATA_SET2_9"))
                                    .tag("DATA_SET1_10", jsonObject.getString("dATA_SET1_10"))
                                    .tag("DATA_SET2_10", jsonObject.getString("dATA_SET2_10"))
                                    .tag("DATA_SET1_11", jsonObject.getString("dATA_SET1_11"))
                                    .tag("DATA_SET2_11", jsonObject.getString("dATA_SET2_11"))
                                    .tag("DATA_SET1_12", jsonObject.getString("dATA_SET1_12"))
                                    .tag("DATA_SET2_12", jsonObject.getString("dATA_SET2_12"))
                                    .tag("DATA_SET1_13", jsonObject.getString("dATA_SET1_13"))
                                    .tag("DATA_SET2_13", jsonObject.getString("dATA_SET2_13"))
                                    .tag("DATA_SET1_14", jsonObject.getString("dATA_SET1_14"))
                                    .tag("DATA_SET2_14", jsonObject.getString("dATA_SET2_14"))
                                    .tag("DATA_SET1_15", jsonObject.getString("dATA_SET1_15"))
                                    .tag("DATA_SET2_15", jsonObject.getString("dATA_SET2_15"))
                                    .tag("DATA_SET1_16", jsonObject.getString("dATA_SET1_16"))
                                    .tag("DATA_SET2_16", jsonObject.getString("dATA_SET2_16"))
                                    .tag("DATA_SET1_17", jsonObject.getString("dATA_SET1_17"))
                                    .tag("DATA_SET2_17", jsonObject.getString("dATA_SET2_17"))
                                    .tag("DATA_SET1_18", jsonObject.getString("dATA_SET1_18"))
                                    .tag("DATA_SET2_18", jsonObject.getString("dATA_SET2_18"))
                                    .tag("DATA_SET1_19", jsonObject.getString("dATA_SET1_19"))
                                    .tag("DATA_SET2_19", jsonObject.getString("dATA_SET2_19"))
                                    .tag("DATA_SET1_20", jsonObject.getString("dATA_SET1_20"))
                                    .tag("DATA_SET2_20", jsonObject.getString("dATA_SET2_20"))
                                    //其他
                                    .tag("weldResult", jsonObject.getString("weldResult"))
                                    .tag("weldConfidence", jsonObject.getString("weldConfidence"))
                                    .tag("defectType", jsonObject.getString("defectType"))
                                    .tag("defectConfidence", jsonObject.getString("defectConfidence"))
                                    .tag("matchType", jsonObject.getString("matchType"))
                                    .tag("writeTime", jsonObject.getString("writeTime"))
                                    .tag("serial", jsonObject.getString("serial"))
                                    .tag("area", jsonObject.getString("area"))
                                    .tag("carModel", jsonObject.getString("carModel"))
                                    .tag("workStation", jsonObject.getString("workStation"))
                                    .tag("spotTag", jsonObject.getString("spotTag"))
                                    .time(System.currentTimeMillis(), TimeUnit.MILLISECONDS).build();
                            batchPoints.point(point);//将point放入batch中

                            Thread.sleep(1);
                        }
                    }
                    influxDbTool.saveInfluxDb(batchPoints, "bosh");
                    // 业务处理成功后，手动提交
                    consumer.commitSync();
                    System.out.println("成功提交" + list.size() + "条,此时的offset为:" + list2.get(99));
                    list.clear();
                    list2.clear();
                    batchPoints.getPoints().clear();


                } else {
                    Thread.sleep(1000);
                }
            } catch (Exception e) {
                //logger.error("bosh存入influxdb失败:"+e.getMessage());
            }
        }
    }

    private void init() {
        Properties props = new Properties();
        //kafka消费的的地址
//        props.put("bootstrap.servers", "172.20.3.81:9092");
        props.put("bootstrap.servers", "127.0.0.1:9092");
        //组名 不同组名可以重复消费
        props.put("group.id", GROUPID);
        //是否自动提交
        props.put("enable.auto.commit", "false");
        //超时时间
        props.put("session.timeout.ms", "30000");
        //一次最大拉取的条数
        props.put("max.poll.records", 100);
//		earliest当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
//		latest
//		当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
//		none
//		topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
        props.put("auto.offset.reset", "earliest");
        //序列化
        props.put("key.deserializer", org.apache.kafka.common.serialization.StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        //订阅主题列表topic
        this.consumer.subscribe(Arrays.asList(topic));

        System.out.println("bosh初始化!");
    }
}
